package io.kiki.sba.registry.remoting.netty;


import io.kiki.sba.registry.api.Callback;
import io.kiki.sba.registry.api.ChannelHandler;
import io.kiki.sba.registry.api.ChannelHandler.HandlerType;
import io.kiki.sba.registry.api.ChannelHandler.InvokeType;
import io.kiki.sba.registry.common.model.store.Url;
import io.kiki.sba.registry.util.CollectionUtils;
import io.kiki.stack.netty.channelManager.ChannelEventProcessor;
import io.kiki.stack.netty.channelManager.ChannelEventType;
import io.kiki.stack.netty.invoker.RequestContext;
import io.kiki.stack.netty.userProcessor.AsyncUserProcessor;
import io.kiki.stack.netty.userProcessor.SyncUserProcessor;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.net.InetSocketAddress;
import java.util.*;
import java.util.concurrent.atomic.AtomicBoolean;


public class Server implements io.kiki.sba.registry.api.Server {
    private static final Logger logger = LoggerFactory.getLogger(Server.class);

    /**
     * accoding server port can not be null
     */
    protected final Url url;

    private final List<ChannelHandler> channelHandlerList;

    private final io.kiki.stack.netty.Server server;

    private final AtomicBoolean isStarted = new AtomicBoolean(false);

    private final AtomicBoolean initHandler = new AtomicBoolean(false);


    public Server(Url url, List<ChannelHandler> channelHandlerList) {
        this.channelHandlerList = channelHandlerList;
        this.url = url;
        this.server = createRpcServer();
    }

    public void configWaterMark(int low, int high) {
        server.initWriteBufferWaterMark(low, high);
        logger.info("config watermark, low={}, high={}", low, high);
    }


    public void startServer() {
        if (isStarted.compareAndSet(false, true)) {
            try {
                initHandler();
                server.start();

            } catch (Exception e) {
                isStarted.set(false);
                logger.error("Start bolt server error!", e);
                throw new RuntimeException("Start bolt server error!", e);
            }
        }
    }

    private void stopServer() {
        if (server != null && isStarted.get()) {
            try {
                server.stop();
            } catch (Exception e) {
                logger.error("Stop bolt server error!", e);
                throw new RuntimeException("Stop bolt server error!", e);
            }
        }
    }


    public void initServer() {
        try {
            initHandler();
        } catch (Exception e) {
            logger.error("Init bolt server error!", e);
            throw new RuntimeException("Init bolt server error!", e);
        }
    }

    protected io.kiki.stack.netty.Server createRpcServer() {
        return new io.kiki.stack.netty.Server(url.getPort(), true);
    }

    private void initHandler() {
        if (initHandler.compareAndSet(false, true)) {
            server.addChannelEventProcessor(ChannelEventType.CONNECT, newConnectionEventProcessor(ChannelEventType.CONNECT));
            server.addChannelEventProcessor(ChannelEventType.CLOSE, newConnectionEventProcessor(ChannelEventType.CLOSE));
            server.addChannelEventProcessor(ChannelEventType.EXCEPTION, newConnectionEventProcessor(ChannelEventType.EXCEPTION));

            registerUserProcessorHandler();
        }
    }

    protected ChannelEventProcessor newConnectionEventProcessor(ChannelEventType type) {
        return new ChannelEventProcessorImpl(type, BoltUtil.getListenerHandlers(channelHandlerList));
    }

    private void registerUserProcessorHandler() {
        for (ChannelHandler channelHandler : channelHandlerList) {
            if (HandlerType.PROCESSER.equals(channelHandler.getHandlerType())) {
                if (InvokeType.SYNC.equals(channelHandler.getInvokeType())) {
                    server.registerUserProcessor(newSyncUserProcessorAdapter(channelHandler));
                } else {
                    server.registerUserProcessor(newAsyncUserProcessorAdapter(channelHandler));
                }
            }
        }
    }

    protected SyncUserProcessor newSyncUserProcessorAdapter(ChannelHandler channelHandler) {
        return new SyncUserProcessorAdapter(channelHandler);
    }

    protected AsyncUserProcessor newAsyncUserProcessorAdapter(ChannelHandler channelHandler) {
        return new AsyncUserProcessorAdapter(channelHandler);
    }

    @Override
    public boolean isOpen() {
        return isStarted.get();
    }

    @Override
    public List<io.kiki.sba.registry.api.Channel> getChannelList() {
        Map<String, List<io.kiki.stack.netty.channelManager.Channel>> map = server.getServerChannelManager().getAll();
        if (map.isEmpty()) {
            return Collections.emptyList();
        }
        List<io.kiki.sba.registry.api.Channel> ret = new ArrayList<>(128);
        for (List<io.kiki.stack.netty.channelManager.Channel> list : map.values()) {
            for (io.kiki.stack.netty.channelManager.Channel conn : list) {
                if (conn.isFine()) {
                    Channel channel = new Channel(conn);
                    ret.add(channel);
                }
            }
        }
        return ret;
    }

    @Override
    public Map<String, io.kiki.sba.registry.api.Channel> selectAvailableChannelsForHostAddress() {
        Map<String, List<io.kiki.sba.registry.api.Channel>> chns = selectAllAvailableChannelsForHostAddress();
        Map<String, io.kiki.sba.registry.api.Channel> ret = new HashMap<>(chns.size());
        for (Map.Entry<String, List<io.kiki.sba.registry.api.Channel>> e : chns.entrySet()) {
            List<io.kiki.sba.registry.api.Channel> list = e.getValue();
            ret.put(e.getKey(), CollectionUtils.getRandom(list));
        }
        return ret;
    }

    @Override
    public Map<String, List<io.kiki.sba.registry.api.Channel>> selectAllAvailableChannelsForHostAddress() {
        Collection<io.kiki.sba.registry.api.Channel> chnList = getChannelList();
        if (chnList.isEmpty()) {
            return Collections.emptyMap();
        }
        Map<String, List<io.kiki.sba.registry.api.Channel>> chns = new HashMap<>(128);
        for (io.kiki.sba.registry.api.Channel chn : chnList) {
            List<io.kiki.sba.registry.api.Channel> list = chns.computeIfAbsent(chn.getRemoteAddress().getAddress().getHostAddress(), k -> new ArrayList<>());
            list.add(chn);
        }
        return chns;
    }

    @Override
    public io.kiki.sba.registry.api.Channel getChannel(InetSocketAddress remoteAddress) {
        Url url = new Url(remoteAddress.getAddress().getHostAddress(), remoteAddress.getPort());
        return getChannel(url);
    }

    @Override
    public io.kiki.sba.registry.api.Channel getChannel(Url url) {
        io.kiki.stack.netty.channelManager.Url key = new io.kiki.stack.netty.channelManager.Url(url.getIp(), url.getPort());
        io.kiki.stack.netty.channelManager.Channel conn = server.getServerChannelManager().get(key.getOriginUrl());
        if (conn == null) {
            return null;
        }
        return new Channel(conn);
    }

    @Override
    public InetSocketAddress getLocalAddress() {
        return new InetSocketAddress(url.getPort());
    }

    @Override
    public void close() {
        stopServer();
    }

    @Override
    public void close(io.kiki.sba.registry.api.Channel channel) {
        if (channel == null) {
            return;
        }
        Channel boltChannel = (Channel) channel;
        io.kiki.stack.netty.channelManager.Channel connection = boltChannel.getChannel();
        if (connection.isFine()) {
            connection.close();
        }
    }

    @Override
    public boolean isClosed() {
        return !isStarted.get();
    }

    @Override
    public Object sendSync(io.kiki.sba.registry.api.Channel channel, Object message, int timeoutMillis) {
        BoltUtil.checkChannelConnected(channel);
        try {
            io.kiki.stack.netty.channelManager.Url boltUrl = BoltUtil.createTargetUrl(channel);
            return server.invokeSync(boltUrl, message, newInvokeContext(message), timeoutMillis);
        } catch (Throwable e) {
            throw BoltUtil.handleException("BoltServer", channel, e, "sendSync");
        }
    }

    @Override
    public void sendCallback(io.kiki.sba.registry.api.Channel channel, Object message, Callback callback, int timeoutMillis) {
        BoltUtil.checkChannelConnected(channel);
        try {
            io.kiki.stack.netty.channelManager.Url boltUrl = BoltUtil.createTargetUrl(channel);
            server.invokeWithCallback(boltUrl, message, newInvokeContext(message), new InvokeCallbackHandler(channel, callback), timeoutMillis);
        } catch (Throwable e) {
            throw BoltUtil.handleException("BoltServer", channel, e, "invokeCallback");
        }
    }

    protected RequestContext newInvokeContext(Object request) {
        return null;
    }

    public io.kiki.stack.netty.Server getRpcServer() {
        return server;
    }

    @Override
    public int getChannelCount() {
        Map<String, List<io.kiki.stack.netty.channelManager.Channel>> map = server.getServerChannelManager().getAll();
        int count = 0;
        for (List<io.kiki.stack.netty.channelManager.Channel> list : map.values()) {
            for (io.kiki.stack.netty.channelManager.Channel conn : list) {
                if (conn.isFine()) {
                    count++;
                }
            }
        }
        return count;
    }
}
